package com.futao.techsharingmq.kafka.basic.simple.consumer;

import com.futao.techsharingmq.common.SecureConfig;
import com.futao.techsharingmq.kafka.basic.simple.config.Config;
import com.futao.techsharingmq.kafka.basic.simple.interceptors.ConsumerInterceptor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

/**
 * @author futao <1185172056@qq.com> <https://github.com/FutaoSmile>
 * @date 2021/3/5
 */
@Slf4j
public class SimpleConsumer {
    public static void main(String[] args) {

        // 消费者配置
        HashMap<String, Object> configs = new HashMap<>();
        // broker地址
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, SecureConfig.MY_STEEL_KAFKA_TEST_URL);
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 消费者ID
        configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "simple-topic-consumer-1");
        // 消费者群组
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "simple-consumer-group");
        // Partition mapping消费者策略
        configs.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RangeAssignor");
        // 自动提交偏移量
        // configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 手动提交偏移量
        configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 提交偏移量的时间间隔
        configs.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");
        configs.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ConsumerInterceptor.class.getName());

        KafkaConsumer<String, String> kafkaConsumer = null;
        try {
            kafkaConsumer = new KafkaConsumer<>(configs);
            HashSet<String> topics = new HashSet<>();
            topics.add(Config.TOPIC);
            // 可正则通配符
            // topics.add("text.*");
            // 订阅主题
            kafkaConsumer.subscribe(topics, new ConsumerRebalanceListener() {
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                    // 在发生再平衡之前用于提交当前消费者在partition的偏移量
                    log.info("在发生再平衡之前用于提交当前消费者在partition的偏移量");
                }

                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    // 在重新分配分区和开始消费分区消息之前被调用
                    log.info("在重新分配分区和开始消费分区消息之前被调用");
                }
            });
            while (true) {
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
                consumerRecords.forEach(consumerRecord -> {
                    log.info("接收到消息:{},{},{},{},{},{},{}", consumerRecord.key(), consumerRecord.headers(), consumerRecord.offset(), consumerRecord.partition(),
                            consumerRecord.topic(), consumerRecord.toString(), consumerRecord.value());
                });
                // 同步提交偏移量-在broker响应之前会一直阻塞-会一直重试
                // kafkaConsumer.commitSync();
                // 异步提交偏移量-不会重试
                kafkaConsumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                        //log.info("偏移量提交结果:{}", map, e);
                    }
                });
            }
        } catch (Exception e) {
            log.error("发生异常", e);
        } finally {
            if (kafkaConsumer != null) {
                // 同步提交偏移量，确保提交成功
                // 通常情况下是异步与同步提交相结合，在最后一次提交的时候进行同步提交，保证偏移量提交成功
                kafkaConsumer.commitSync();
                kafkaConsumer.close();
            }
        }
    }
}
